package com.education.realtime.app.dwd;

import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdExam extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdExam().init(2001,"DwdExamPaperse");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"DwdExamPaperse");
        Table test_exam = tEnv.sqlQuery(
                " select " +
                        "data['id'] id , " +
                        "data['paper_id'] paper_id, " +
                        "data['user_id'] user_id, " +
                        "data['score'] score, " +
                        "data['duration_sec'] duration_sec, " +
                        "data['create_time'] create_time, " +
                        "data['submit_time'] submit_time, " +
                        "data['update_time'] update_time, " +
                        "data['deleted'] deleted, " +
                        "ts " +
                        "from ods_db " +
                        "where `table`= 'test_exam' " +
                        "and `type`='insert' ");
        tEnv.createTemporaryView("test_exam",test_exam);
//        Table test_paper = tEnv.sqlQuery("select " +
//                "data['id'] id, " +
//                "data['paper_title'] paper_title, " +
//                "data['course_id'] course_id, " +
//                "data['create_time'] create_time, " +
//                "data['update_time'] update_time, " +
//                "data['publisher_id'] publisher_id, " +
//                "data['deleted'] deleted, " +
//                "ts " +
//                "from ods_db " +
//                "where `table`='test_paper' " +
//                "and `type`='bootstrap-insert'");

        //join
//        Table table = tEnv.sqlQuery("select " +
//                "t1.id, " +
//                "t1.paper_id, " +
//                "t1.user_id, " +
//                "t1.score, " +
//                "t1.duration_sec, " +
//                "t1.create_time exam_create_time, " +
//                "t1.submit_time exam_update_time, " +
//                "t1.update_time exam_update_time, " +
//                "t1.deleted exam_deleted, " +
//                "t2.paper_title, " +
//                "t2.course_id, " +
//                "t2.create_time paper_create_time, " +
//                "t2.update_time paper_update_time, " +
//                "t2.publisher_id, " +
//                "t2.deleted paper_deleted" +
//                "t1.ts exam_ts, " +
//                "t2.ts paper_ts" +
//                "from test_exam t1 " +
//                "join test_paper t2 on t1.paper_id=t2.id ");
//        //写入
//        tEnv.executeSql("create table dwd_exam_paper("  +
//                "id, " +
//                "paper_id, " +
//                "user_id, " +
//                "score, " +
//                "duration_sec, " +
//                "exam_create_time, " +
//                "exam_update_time, " +
//                "exam_update_time, " +
//                "deleted exam_deleted, " +
//                "paper_title, " +
//                "course_id, " +
//                "create_time paper_create_time, " +
//                "update_time paper_update_time, " +
//                "publisher_id, " +
//                "deleted paper_deleted" +
//                "ts exam_ts, " +
//                "ts paper_ts" +
//                ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_EXAM_PAPER));
                tEnv.executeSql("create table dwd_exam("  +
                        "id string, " +
                        "paper_id string, " +
                        "user_id string, " +
                        "score string, " +
                        "duration_sec string, " +
                        "create_time string, " +
                        "submit_time string, " +
                        "update_time string, " +
                        "deleted string, " +
                        "ts string" +
                        ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_EXAM));
                test_exam.executeInsert("dwd_exam");

    }
}